// PricingStrategyMapper.java
package com.taxi.analysis.pricing;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 智能定价策略分析Mapper
 * 功能：使用高级聚合技术分析定价策略
 * 技术：高级聚合技术（多维度聚合、嵌套聚合、条件聚合、滑动窗口聚合）
 */
public class PricingStrategyMapper extends Mapper<LongWritable, Text, Text, Text> {

    private Text outputKey = new Text();
    private Text outputValue = new Text();

    @Override
    public void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {

        if (key.get() == 0) return;

        try {
            String line = value.toString().trim();
            if (line.isEmpty()) return;

            String[] fields = parseCSVLine(line);

            if (fields.length >= 33) {
                int pickupHour = parseInt(fields[0]);
                int pickupWeekday = parseInt(fields[1]);
                String pickupDate = fields[2].trim();
                int pickupMonth = parseInt(fields[3]);
                int pickupDay = parseInt(fields[4]);
                String pickupLocationId = fields[5].trim();
                String dropoffLocationId = fields[6].trim();
                double tripDistance = parseDouble(fields[7]);
                double tripDuration = parseDouble(fields[8]);
                double avgSpeed = parseDouble(fields[9]);
                double fareAmount = parseDouble(fields[10]);
                double tipAmount = parseDouble(fields[11]);
                double tollsAmount = parseDouble(fields[12]);
                double extra = parseDouble(fields[13]);
                double totalAmount = parseDouble(fields[14]);
                double baseFare = parseDouble(fields[15]);
                double farePerMile = parseDouble(fields[16]);
                double farePerMinute = parseDouble(fields[17]);
                double tipRate = parseDouble(fields[18]);
                double totalRevenueRate = parseDouble(fields[19]);
                String timePeriod = fields[20].trim();
                String dayType = fields[21].trim();
                String distanceCategory = fields[22].trim();
                String speedCategory = fields[23].trim();
                String pricingTier = fields[24].trim();
                double surgeMultiplier = parseDouble(fields[25]);
                double suggestedFare = parseDouble(fields[26]);
                double pricingEfficiency = parseDouble(fields[27]);
                double routeAvgFare = parseDouble(fields[28]);
                double priceCompetitiveness = parseDouble(fields[29]);
                String competitivenessLevel = fields[30].trim();
                double timeValue = parseDouble(fields[31]);
                double distanceValue = parseDouble(fields[32]);
                double valueScore = parseDouble(fields[33]);
                String valueLevel = fields[34].trim();
                String marketSegment = fields[35].trim();
                String customerType = fields[36].trim();
                int passengerCount = parseInt(fields[37]);

                if (isValidData(pickupLocationId, dropoffLocationId)) {

                    // 高级聚合技术1: 多维度聚合
                    processMultiDimensionalAggregation(pickupHour, pickupWeekday, pickupMonth,
                            timePeriod, dayType, distanceCategory, pricingTier,
                            fareAmount, tipAmount, totalAmount, tripDistance,
                            tripDuration, farePerMile, tipRate, valueScore, context);

                    // 高级聚合技术2: 嵌套聚合
                    processNestedAggregation(pickupLocationId, dropoffLocationId, timePeriod,
                            dayType, pricingTier, customerType, fareAmount,
                            tipAmount, totalAmount, tripDistance, farePerMile,
                            surgeMultiplier, pricingEfficiency, valueScore, context);

                    // 高级聚合技术3: 条件聚合
                    processConditionalAggregation(pickupHour, timePeriod, dayType, distanceCategory,
                            speedCategory, pricingTier, competitivenessLevel,
                            valueLevel, customerType, fareAmount, tipAmount,
                            totalAmount, tripDistance, farePerMile, tipRate,
                            surgeMultiplier, pricingEfficiency, context);

                    // 高级聚合技术4: 滑动窗口聚合
                    processSlidingWindowAggregation(pickupDate, pickupHour, pickupLocationId,
                            timePeriod, dayType, fareAmount, tipAmount,
                            totalAmount, tripDistance, farePerMile,
                            surgeMultiplier, valueScore, context);

                    // 高级聚合技术5: 分层聚合
                    processHierarchicalAggregation(pickupLocationId, dropoffLocationId, timePeriod,
                            dayType, distanceCategory, pricingTier, customerType,
                            fareAmount, tipAmount, totalAmount, tripDistance,
                            farePerMile, tipRate, valueScore, context);
                }
            }

        } catch (Exception e) {
            System.err.println("处理行时出错: " + value.toString());
        }
    }

    /**
     * 多维度聚合：按多个维度同时聚合
     */
    private void processMultiDimensionalAggregation(int pickupHour, int pickupWeekday, int pickupMonth,
                                                    String timePeriod, String dayType, String distanceCategory,
                                                    String pricingTier, double fareAmount, double tipAmount,
                                                    double totalAmount, double tripDistance, double tripDuration,
                                                    double farePerMile, double tipRate, double valueScore,
                                                    Context context) throws IOException, InterruptedException {

        // 时间维度聚合
        String timeKey = String.format("MULTI_TIME_%s_%s_%d_%d_%d", timePeriod, dayType, pickupHour, pickupWeekday, pickupMonth);
        String timeRecord = createMultiDimensionalRecord("TIME", fareAmount, tipAmount, totalAmount,
                tripDistance, tripDuration, farePerMile, tipRate, valueScore);
        outputKey.set(timeKey);
        outputValue.set("MULTI_DIMENSIONAL|" + timeRecord);
        context.write(outputKey, outputValue);

        // 距离-定价维度聚合
        String distancePriceKey = String.format("MULTI_DISTANCE_PRICE_%s_%s", distanceCategory, pricingTier);
        String distancePriceRecord = createMultiDimensionalRecord("DISTANCE_PRICE", fareAmount, tipAmount,
                totalAmount, tripDistance, tripDuration,
                farePerMile, tipRate, valueScore);
        outputKey.set(distancePriceKey);
        outputValue.set("MULTI_DIMENSIONAL|" + distancePriceRecord);
        context.write(outputKey, outputValue);

        // 时间-距离-定价三维聚合
        String tripleKey = String.format("MULTI_TRIPLE_%s_%s_%s", timePeriod, distanceCategory, pricingTier);
        String tripleRecord = createMultiDimensionalRecord("TRIPLE", fareAmount, tipAmount, totalAmount,
                tripDistance, tripDuration, farePerMile, tipRate, valueScore);
        outputKey.set(tripleKey);
        outputValue.set("MULTI_DIMENSIONAL|" + tripleRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 嵌套聚合：先按一个维度聚合，再在结果基础上聚合
     */
    private void processNestedAggregation(String pickupLocationId, String dropoffLocationId, String timePeriod,
                                          String dayType, String pricingTier, String customerType,
                                          double fareAmount, double tipAmount, double totalAmount,
                                          double tripDistance, double farePerMile, double surgeMultiplier,
                                          double pricingEfficiency, double valueScore, Context context)
            throws IOException, InterruptedException {

        // 第一层：按路线聚合
        String routeKey = String.format("NESTED_L1_ROUTE_%s_%s", pickupLocationId, dropoffLocationId);
        String routeRecord = createNestedRecord("ROUTE", timePeriod, dayType, pricingTier, customerType,
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, surgeMultiplier, pricingEfficiency, valueScore);
        outputKey.set(routeKey);
        outputValue.set("NESTED_L1|" + routeRecord);
        context.write(outputKey, outputValue);

        // 第二层：按时间-客户类型聚合
        String timeCustomerKey = String.format("NESTED_L2_TIME_CUSTOMER_%s_%s_%s", timePeriod, dayType, customerType);
        String timeCustomerRecord = createNestedRecord("TIME_CUSTOMER", pickupLocationId, dropoffLocationId,
                pricingTier, "", fareAmount, tipAmount, totalAmount,
                tripDistance, farePerMile, surgeMultiplier,
                pricingEfficiency, valueScore);
        outputKey.set(timeCustomerKey);
        outputValue.set("NESTED_L2|" + timeCustomerRecord);
        context.write(outputKey, outputValue);

        // 第三层：按定价层级聚合
        String pricingKey = String.format("NESTED_L3_PRICING_%s", pricingTier);
        String pricingRecord = createNestedRecord("PRICING", timePeriod, dayType, customerType, "",
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, surgeMultiplier, pricingEfficiency, valueScore);
        outputKey.set(pricingKey);
        outputValue.set("NESTED_L3|" + pricingRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 条件聚合：基于特定条件进行聚合
     */
    private void processConditionalAggregation(int pickupHour, String timePeriod, String dayType,
                                               String distanceCategory, String speedCategory, String pricingTier,
                                               String competitivenessLevel, String valueLevel, String customerType,
                                               double fareAmount, double tipAmount, double totalAmount,
                                               double tripDistance, double farePerMile, double tipRate,
                                               double surgeMultiplier, double pricingEfficiency, Context context)
            throws IOException, InterruptedException {

        // 条件1：高峰时段高价值行程
        if ((timePeriod.equals("早高峰") || timePeriod.equals("晚高峰")) && valueLevel.equals("高价值")) {
            String peakValueKey = "CONDITIONAL_PEAK_HIGH_VALUE";
            String peakValueRecord = createConditionalRecord("PEAK_HIGH_VALUE", pickupHour, timePeriod,
                    dayType, distanceCategory, pricingTier, fareAmount,
                    tipAmount, totalAmount, tripDistance, farePerMile,
                    tipRate, surgeMultiplier, pricingEfficiency);
            outputKey.set(peakValueKey);
            outputValue.set("CONDITIONAL|" + peakValueRecord);
            context.write(outputKey, outputValue);
        }

        // 条件2：高竞争力长途行程
        if (competitivenessLevel.equals("高竞争力") && (distanceCategory.equals("长途") || distanceCategory.equals("超长途"))) {
            String competitiveLongKey = "CONDITIONAL_COMPETITIVE_LONG";
            String competitiveLongRecord = createConditionalRecord("COMPETITIVE_LONG", pickupHour, timePeriod,
                    dayType, distanceCategory, pricingTier, fareAmount,
                    tipAmount, totalAmount, tripDistance, farePerMile,
                    tipRate, surgeMultiplier, pricingEfficiency);
            outputKey.set(competitiveLongKey);
            outputValue.set("CONDITIONAL|" + competitiveLongRecord);
            context.write(outputKey, outputValue);
        }

        // 条件3：高端客户拥堵时段
        if (customerType.equals("高端客户") && speedCategory.equals("拥堵")) {
            String premiumCongestionKey = "CONDITIONAL_PREMIUM_CONGESTION";
            String premiumCongestionRecord = createConditionalRecord("PREMIUM_CONGESTION", pickupHour,
                    timePeriod, dayType, distanceCategory,
                    pricingTier, fareAmount, tipAmount,
                    totalAmount, tripDistance, farePerMile,
                    tipRate, surgeMultiplier, pricingEfficiency);
            outputKey.set(premiumCongestionKey);
            outputValue.set("CONDITIONAL|" + premiumCongestionRecord);
            context.write(outputKey, outputValue);
        }

        // 条件4：工作日经济型定价
        if (dayType.equals("工作日") && pricingTier.equals("经济型")) {
            String weekdayEconomyKey = "CONDITIONAL_WEEKDAY_ECONOMY";
            String weekdayEconomyRecord = createConditionalRecord("WEEKDAY_ECONOMY", pickupHour, timePeriod,
                    dayType, distanceCategory, pricingTier, fareAmount,
                    tipAmount, totalAmount, tripDistance, farePerMile,
                    tipRate, surgeMultiplier, pricingEfficiency);
            outputKey.set(weekdayEconomyKey);
            outputValue.set("CONDITIONAL|" + weekdayEconomyRecord);
            context.write(outputKey, outputValue);
        }
    }

    /**
     * 滑动窗口聚合：基于时间窗口进行聚合
     */
    private void processSlidingWindowAggregation(String pickupDate, int pickupHour, String pickupLocationId,
                                                 String timePeriod, String dayType, double fareAmount,
                                                 double tipAmount, double totalAmount, double tripDistance,
                                                 double farePerMile, double surgeMultiplier, double valueScore,
                                                 Context context) throws IOException, InterruptedException {

        // 小时窗口（当前小时及前后1小时）
        for (int hourOffset = -1; hourOffset <= 1; hourOffset++) {
            int windowHour = (pickupHour + hourOffset + 24) % 24;
            String hourWindowKey = String.format("SLIDING_HOUR_%s_%d_%s", pickupDate, windowHour, pickupLocationId);
            String hourWindowRecord = createSlidingWindowRecord("HOUR", pickupDate, pickupHour, timePeriod,
                    dayType, fareAmount, tipAmount, totalAmount,
                    tripDistance, farePerMile, surgeMultiplier, valueScore);
            outputKey.set(hourWindowKey);
            outputValue.set("SLIDING_WINDOW|" + hourWindowRecord);
            context.write(outputKey, outputValue);
        }

        // 时段窗口
        String periodWindowKey = String.format("SLIDING_PERIOD_%s_%s_%s", pickupDate, timePeriod, pickupLocationId);
        String periodWindowRecord = createSlidingWindowRecord("PERIOD", pickupDate, pickupHour, timePeriod,
                dayType, fareAmount, tipAmount, totalAmount,
                tripDistance, farePerMile, surgeMultiplier, valueScore);
        outputKey.set(periodWindowKey);
        outputValue.set("SLIDING_WINDOW|" + periodWindowRecord);
        context.write(outputKey, outputValue);

        // 日类型窗口
        String dayTypeWindowKey = String.format("SLIDING_DAYTYPE_%s_%s_%s", pickupDate, dayType, pickupLocationId);
        String dayTypeWindowRecord = createSlidingWindowRecord("DAYTYPE", pickupDate, pickupHour, timePeriod,
                dayType, fareAmount, tipAmount, totalAmount,
                tripDistance, farePerMile, surgeMultiplier, valueScore);
        outputKey.set(dayTypeWindowKey);
        outputValue.set("SLIDING_WINDOW|" + dayTypeWindowRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 分层聚合：按层次结构进行聚合
     */
    private void processHierarchicalAggregation(String pickupLocationId, String dropoffLocationId, String timePeriod,
                                                String dayType, String distanceCategory, String pricingTier,
                                                String customerType, double fareAmount, double tipAmount,
                                                double totalAmount, double tripDistance, double farePerMile,
                                                double tipRate, double valueScore, Context context)
            throws IOException, InterruptedException {

        // 层次1：全局聚合
        String globalKey = "HIERARCHICAL_GLOBAL";
        String globalRecord = createHierarchicalRecord("GLOBAL", "ALL", "ALL", "ALL", "ALL",
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, tipRate, valueScore);
        outputKey.set(globalKey);
        outputValue.set("HIERARCHICAL|" + globalRecord);
        context.write(outputKey, outputValue);

        // 层次2：按时间类型聚合
        String timeTypeKey = String.format("HIERARCHICAL_TIME_TYPE_%s_%s", timePeriod, dayType);
        String timeTypeRecord = createHierarchicalRecord("TIME_TYPE", timePeriod, dayType, "ALL", "ALL",
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, tipRate, valueScore);
        outputKey.set(timeTypeKey);
        outputValue.set("HIERARCHICAL|" + timeTypeRecord);
        context.write(outputKey, outputValue);

        // 层次3：按距离-定价聚合
        String distancePricingKey = String.format("HIERARCHICAL_DISTANCE_PRICING_%s_%s", distanceCategory, pricingTier);
        String distancePricingRecord = createHierarchicalRecord("DISTANCE_PRICING", distanceCategory,
                pricingTier, "ALL", "ALL", fareAmount, tipAmount,
                totalAmount, tripDistance, farePerMile, tipRate, valueScore);
        outputKey.set(distancePricingKey);
        outputValue.set("HIERARCHICAL|" + distancePricingRecord);
        context.write(outputKey, outputValue);

        // 层次4：按客户类型聚合
        String customerKey = String.format("HIERARCHICAL_CUSTOMER_%s", customerType);
        String customerRecord = createHierarchicalRecord("CUSTOMER", customerType, "ALL", "ALL", "ALL",
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, tipRate, valueScore);
        outputKey.set(customerKey);
        outputValue.set("HIERARCHICAL|" + customerRecord);
        context.write(outputKey, outputValue);

        // 层次5：按路线聚合
        String routeKey = String.format("HIERARCHICAL_ROUTE_%s_%s", pickupLocationId, dropoffLocationId);
        String routeRecord = createHierarchicalRecord("ROUTE", pickupLocationId, dropoffLocationId, "ALL", "ALL",
                fareAmount, tipAmount, totalAmount, tripDistance,
                farePerMile, tipRate, valueScore);
        outputKey.set(routeKey);
        outputValue.set("HIERARCHICAL|" + routeRecord);
        context.write(outputKey, outputValue);
    }

    /**
     * 创建多维度聚合记录
     */
    private String createMultiDimensionalRecord(String dimension, double fareAmount, double tipAmount,
                                                double totalAmount, double tripDistance, double tripDuration,
                                                double farePerMile, double tipRate, double valueScore) {
        return String.format("%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%.3f|%.2f",
                dimension, fareAmount, tipAmount, totalAmount, tripDistance,
                tripDuration, farePerMile, tipRate, valueScore);
    }

    /**
     * 创建嵌套聚合记录
     */
    private String createNestedRecord(String level, String dim1, String dim2, String dim3, String dim4,
                                      double fareAmount, double tipAmount, double totalAmount,
                                      double tripDistance, double farePerMile, double surgeMultiplier,
                                      double pricingEfficiency, double valueScore) {
        return String.format("%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%.3f|%.2f",
                level, dim1, dim2, dim3, dim4, fareAmount, tipAmount, totalAmount,
                tripDistance, farePerMile, surgeMultiplier, pricingEfficiency, valueScore);
    }

    /**
     * 创建条件聚合记录
     */
    private String createConditionalRecord(String condition, int pickupHour, String timePeriod,
                                           String dayType, String distanceCategory, String pricingTier,
                                           double fareAmount, double tipAmount, double totalAmount,
                                           double tripDistance, double farePerMile, double tipRate,
                                           double surgeMultiplier, double pricingEfficiency) {
        return String.format("%s|%d|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.3f|%.2f|%.3f",
                condition, pickupHour, timePeriod, dayType, distanceCategory, pricingTier,
                fareAmount, tipAmount, totalAmount, tripDistance, farePerMile, tipRate,
                surgeMultiplier, pricingEfficiency);
    }

    /**
     * 创建滑动窗口聚合记录
     */
    private String createSlidingWindowRecord(String windowType, String pickupDate, int pickupHour,
                                             String timePeriod, String dayType, double fareAmount,
                                             double tipAmount, double totalAmount, double tripDistance,
                                             double farePerMile, double surgeMultiplier, double valueScore) {
        return String.format("%s|%s|%d|%s|%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f|%.2f",
                windowType, pickupDate, pickupHour, timePeriod, dayType, fareAmount,
                tipAmount, totalAmount, tripDistance, farePerMile, surgeMultiplier, valueScore);
    }

    /**
     * 创建分层聚合记录
     */
    private String createHierarchicalRecord(String level, String dim1, String dim2, String dim3, String dim4,
                                            double fareAmount, double tipAmount, double totalAmount,
                                            double tripDistance, double farePerMile, double tipRate, double valueScore) {
        return String.format("%s|%s|%s|%s|%s|%.2f|%.2f|%.2f|%.2f|%.2f|%.3f|%.2f",
                level, dim1, dim2, dim3, dim4, fareAmount, tipAmount, totalAmount,
                tripDistance, farePerMile, tipRate, valueScore);
    }

    private boolean isValidData(String pickupLocationId, String dropoffLocationId) {
        return !pickupLocationId.isEmpty() && !dropoffLocationId.isEmpty() &&
                !pickupLocationId.equals("null") && !dropoffLocationId.equals("null");
    }

    private String[] parseCSVLine(String line) {
        if (line.contains("\"")) {
            return parseQuotedCSV(line);
        } else {
            return line.split(",");
        }
    }

    private String[] parseQuotedCSV(String line) {
        List<String> fields = new ArrayList<>();
        boolean inQuotes = false;
        StringBuilder currentField = new StringBuilder();

        for (int i = 0; i < line.length(); i++) {
            char c = line.charAt(i);

            if (c == '"') {
                inQuotes = !inQuotes;
            } else if (c == ',' && !inQuotes) {
                fields.add(currentField.toString());
                currentField = new StringBuilder();
            } else {
                currentField.append(c);
            }
        }
        fields.add(currentField.toString());

        return fields.toArray(new String[0]);
    }

    private double parseDouble(String str) {
        try {
            return Double.parseDouble(str.trim());
        } catch (NumberFormatException e) {
            return 0.0;
        }
    }

    private int parseInt(String str) {
        try {
            return Integer.parseInt(str.trim());
        } catch (NumberFormatException e) {
            return 0;
        }
    }
}
